package produce

import (
	"context"
	"data-process/logger"
	"database/sql"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
)

type MysqlDbProduce struct {
	CommonInfo
	sql  string
	link string
}

func NewMySqlDbProduce(sql string, link string) *MysqlDbProduce {
	return &MysqlDbProduce{
		sql:  sql,
		link: link,
	}
}

func (m *MysqlDbProduce) buildSource() (*sql.Conn, error) {
	linkUrl := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", m.username, m.password, m.ip, m.port, m.database)
	db, err := sql.Open("mysql", linkUrl)
	if err != nil {
		return nil, err
	}
	db.SetConnMaxLifetime(100)
	db.SetMaxOpenConns(5)
	if err := db.Ping(); err != nil {
		return nil, err
	}
	conn, err := db.Conn(context.TODO())
	return conn, nil
}

func (m *MysqlDbProduce) readData(data chan<- map[string]interface{}, size int) {
	conn, err := m.buildSource()
	if err != nil {

	}
	rows, err := conn.QueryContext(context.TODO(), fmt.Sprintf("%s limit %d", m.sql, size))
	if err != nil {

	}
	columns, err := rows.Columns()
	if err != nil {

	}
	logger.Infof("列信息：%v", columns)
	values := make([]sql.RawBytes, len(columns))
	scanArgs := make([]interface{}, len(values))
	for i := range values {
		scanArgs[i] = &values[i]
	}

	for rows.Next() {
		itemData := make(ItemDataMap, len(columns))

		_ = rows.Scan(scanArgs...)

		var value string
		for i, col := range values {
			if col == nil {
				value = "NULL"
			} else {
				value = string(col)
			}
			itemData[columns[i]] = value
		}

		//for idx, item := range columns {
		//	fmt.Println(cache[idx])
		//	itemData[item] = *cache[idx].(*interface{})
		//}
		data <- itemData
	}

}
